package org.ngbx.demo.flink.clickevent.statistics.functions;

import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.ngbx.demo.flink.clickevent.statistics.event.UserClickStatisticsPeriodicallyCompletedEvent;

public class UserClickStatisticsPeriodicallyCompletedEventProcessor
        extends ProcessWindowFunction<Long, UserClickStatisticsPeriodicallyCompletedEvent, Long, TimeWindow> {

    @Override
    public void process(
            Long userId,
            ProcessWindowFunction<Long, UserClickStatisticsPeriodicallyCompletedEvent, Long, TimeWindow>.Context context,
            Iterable<Long> elements,
            Collector<UserClickStatisticsPeriodicallyCompletedEvent> out
    ) throws Exception {
        Long clicks = elements.iterator().next();

        out.collect(new UserClickStatisticsPeriodicallyCompletedEvent(
                context.window().getStart(),
                context.window().getEnd(),
                userId, clicks)
        );
    }
}